package com.asap.demo.sourcefunc;

import com.alibaba.fastjson.JSON;
import com.asap.demo.model.RuleBean;
import com.asap.rule.operator.StringIn;
import com.asap.rule.orm.DbFetcher;
import com.asap.rule.util.Constants;
import com.asap.rule.util.RuleReader;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

public class MysqlSourceFunction1 extends RichSourceFunction<List<String>> {

	private static final Logger logger = LoggerFactory.getLogger(MysqlSourceFunction1.class);

	private PreparedStatement ps;
	private Connection conn;
	DbFetcher dbFetcher;

	public MysqlSourceFunction1(DbFetcher dbFetcher) {
		this.dbFetcher = dbFetcher;
	}


	@Override
	public void run(SourceContext<List<String>> ctx) throws Exception {
		List<String> list = new ArrayList<>();
		list.add("0");
		list.addAll(RuleReader.readRules(dbFetcher));
		logger.info("list size:" + list.size());
		ctx.collect(list);
		dbFetcher.updateConfigUpdateLog();
		while (true) {
			if (dbFetcher.getConfigUpdateLog() == 1) {
				logger.info("MysqlSourceFunction1 config update..............");
				list.clear();
				list.add("0");
				list.addAll(RuleReader.readRules(dbFetcher));
				ctx.collect(list);
				dbFetcher.updateConfigUpdateLog();
			}else{
				list.set(0,"1");
				ctx.collect(list);
			}
			Thread.sleep(10000);
		}


//		ResultSet rs = ps.executeQuery();
//		while (rs.next()) {
//			String configJson = rs.getString("CONFIG_JSON");
//			RuleBean ruleBean = JSON.parseObject(configJson, RuleBean.class);
//			logger.info("rule: {}", ruleBean.toString());
//			ctx.collect(configJson);
//		}
	}

	@Override
	public void cancel() {
	}

	@Override
	public void close() throws Exception {
		super.close();
//		if (conn != null)
//			conn.close();
//		if (ps != null)
//			ps.close();
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
//		this.conn = getConn();
//		logger.info("3333333:",this.conn);
//		String sql = "select CONFIG_JSON from RE_RULE";
//		this.ps = this.conn.prepareStatement(sql);
	}

	@Override
	protected Object clone() throws CloneNotSupportedException {
		return super.clone();
	}

	public static Connection getConn() {
		Connection conn = null;
		try {
			Class.forName("com.mysql.jdbc.Driver");
			conn = DriverManager.getConnection(
					"jdbc:mysql://192.168.1.238:3306/SSA?useUnicode=true&characterEncoding=utf-8",
					"root",
					"1qazXSW@3edc");
		} catch (ClassNotFoundException | SQLException e) {
			logger.error(e.getMessage(),e);
			System.out.println(e);
		}
		return conn;
	}

}
